package server_agent

import (
	"encoding/json"
	"lark/conf"
	"lark/consts"
	"lark/proto"
	"lark/util"
	"net"
	"os"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	etcdclient "github.com/coreos/etcd/clientv3"
	"github.com/pkg/errors"
	"github.com/zeast/logs"
)

type ProxyManager struct {
	sync.Mutex
	lis           net.Listener
	parser        *proto.LarkProtoV2
	proxyMap      map[string]ServerInterface
	status        uint32
	newServerFunc func(ServiceRegistInfo) (ServerInterface, error)
	etcdclients   map[string]*etcdclient.Client
}

type ServiceRegistInfo struct {
	Ip            string `json:"ip"`
	Port          int    `json:"port"`
	ServiceName   string `json:"service"`
	WorkNum       int    `json:"worker_num"`
	CgiPass       string `json:"cgi_pass"`
	KeepAliveTime time.Time
	Kind          string `json:"kind"` //用于部分逻辑的区别处理，现在有重试逻辑
}

func (info ServiceRegistInfo) larkString() string {
	return "lark.address." + info.ServiceName + "." + strconv.Itoa(info.WorkNum) + "." + info.Ip + ":" + strconv.Itoa(info.Port)
}

func (info ServiceRegistInfo) mcpString() string {
	return "service.address." + info.ServiceName + "." + info.Ip + ":" + strconv.Itoa(info.Port)
}

func (info ServiceRegistInfo) String() string {
	return strings.Join([]string{info.ServiceName, info.CgiPass, strconv.Itoa(info.WorkNum)}, "-")
}

var DefaultProxyManager *ProxyManager

func Init() error {
	if err := initLogger(); err != nil {
		return err
	}

	pm, err := newProxyManager(conf.ServerListenStr, NewService)
	if err != nil {
		return err
	}

	DefaultProxyManager = pm
	go DefaultProxyManager.Run()
	return nil
}

//管理所有的代理，串行处理来自控制端口的消息
func newProxyManager(listenStr string, newServerFunc func(ServiceRegistInfo) (ServerInterface, error)) (*ProxyManager, error) {
	strs := strings.Split(listenStr, "://")
	if len(strs) != 2 {
		return nil, errors.Errorf("server listen str is error:%s", listenStr)
	}

	network := strs[0]
	address := strs[1]

	if network == "unix" {
		err := util.EnsureDir(address, 0755)
		if err != nil {
			return nil, errors.Cause(err)
		}
	}

	l, err := net.Listen(network, address)
	if err != nil {
		return nil, errors.Cause(err)
	}

	if network == "unix" {
		if err = os.Chmod(address, 0766); err != nil {
			return nil, errors.Cause(err)
		}
	}

	return &ProxyManager{
		lis:           l,
		status:        newFlag,
		proxyMap:      make(map[string]ServerInterface),
		newServerFunc: newServerFunc,
		etcdclients:   make(map[string]*etcdclient.Client),
	}, nil
}

//串行处理
func (pm *ProxyManager) Run() {
	if !atomic.CompareAndSwapUint32(&pm.status, newFlag, runningFlag) {
		return
	}

	go pm.autoCheckExpireService()
	go pm.autoCheckBlockingTask()

	logger.Infof("proxy manager start")
	for {
		conn, err := pm.lis.Accept()
		if err != nil {
			if atomic.LoadUint32(&pm.status) == closedFlag {
				return
			}

			if ne, ok := err.(net.Error); ok && ne.Temporary() {
				time.Sleep(time.Millisecond * 100)
			} else {
				time.Sleep(time.Second)
			}
			logger.Error(err)
			continue
		}

		pm.handle(conn)
	}
}

func (pm *ProxyManager) handle(conn net.Conn) {
	defer util.PrintPanicStack()
	defer conn.Close()

	pm.Lock()
	defer pm.Unlock()

	parser := proto.NewlarkProtoV2(conn, conf.ServerAgentReadLocalTimeout(), conf.ServerAgentWriteLocalTimeout())
	tag, payload, err := parser.Read()
	if err != nil {
		logs.Error(err)
		return
	}

	logger.Debugf("proxy manager receive message. tag:%d, value:%s ", tag, payload)
	switch tag {
	case consts.Ps2lsmTagRegistReq:
		err = pm.handleRegistReq(payload)
		if err != nil {
			parser.Write(consts.Lsm2psTagRegistRespERR, []byte(err.Error()))
		} else {
			parser.Write(consts.Lsm2psTagRegistRespOK, []byte("OK"))
		}

	case consts.Ps2lsmTagUnRegistReq:
		err = pm.handleUnRegistReq(payload)
		if err != nil {
			parser.Write(consts.Lsm2psTagUnRegistRespERR, []byte(err.Error()))
		} else {
			parser.Write(consts.Lsm2psTagUnRegistRespOK, []byte("OK"))
		}

	default:
		logger.Errorf("proxy manager. can't find handle for tag %d", tag)
	}

	if err != nil {
		logger.Errorf("%+v", err)

		//清空到 etcd 的链接
		for k, cli := range pm.etcdclients {
			cli.Close()
			delete(pm.etcdclients, k)
		}
	}

	return
}

func (pm *ProxyManager) handleRegistReq(payload []byte) error {
	var err error
	var registInfoList []ServiceRegistInfo
	var etcdServerList map[string][]string

	segments := proto.SplitPayload(payload)
	err = json.Unmarshal(segments[0], &registInfoList)
	if err != nil {
		err = errors.WithStack(err)
		return err
	}

	err = json.Unmarshal(segments[1], &etcdServerList)
	if err != nil {
		err = errors.WithStack(err)
		return err
	}

	num := len(etcdServerList) * len(registInfoList) * 2
	var retc = make(chan error, num)
	var rerr error
	var etcderr error

	for _, info := range registInfoList {
		if _, ok := pm.proxyMap[info.String()]; !ok {
			p, err := pm.newServerFunc(info)
			if err != nil {
				if rerr == nil {
					rerr = err
				} else {
					rerr = errors.WithMessage(rerr, err.Error())
				}

				continue
			} else {
				go p.Run()
				pm.proxyMap[info.String()] = p
				logger.Debugf("add server ok, %s", info.String())
			}
		} else {
			pm.proxyMap[info.String()].KeepAlive()
		}

		for _, endpoints := range etcdServerList {
			if etcderr != nil {
				continue
			}

			k := strings.Join(endpoints, ",")
			cli, ok := pm.etcdclients[k]
			if !ok {
				cli, err = etcdclient.New(etcdclient.Config{
					Endpoints:            endpoints,
					AutoSyncInterval:     conf.EtcdSyncEndpointsTime,
					DialTimeout:          conf.EtcdDialTimeout,
					DialKeepAliveTime:    conf.EtcdDialKeepAliveTime,
					DialKeepAliveTimeout: conf.EtcdDialKeepAliveTimeout,
				})

				if err != nil {
					etcderr = err
					continue
				}

				pm.etcdclients[k] = cli
			}

			go KeepAlive(cli, info.larkString(), retc)
			go KeepAlive(cli, info.mcpString(), retc)
		}
	}

	if etcderr != nil {
		return etcderr
	}

	for i := 0; i < num; i++ {
		select {
		case err := <-retc:
			if err != nil {
				if rerr == nil {
					rerr = err
				} else {
					rerr = errors.WithMessage(rerr, err.Error())
				}
			}

		case <-time.After(time.Second * 10):
			m := "keepalive wait response timeout"
			if rerr == nil {
				rerr = errors.New(m)
			} else {
				rerr = errors.WithMessage(rerr, m)
			}
		}
	}

	return rerr
}

func (pm *ProxyManager) handleUnRegistReq(payload []byte) error {
	var err error
	var registInfoList []ServiceRegistInfo
	var etcdServerList map[string][]string

	segments := proto.SplitPayload(payload)
	err = json.Unmarshal(segments[0], &registInfoList)
	if err != nil {
		err = errors.WithStack(err)
		return err
	}

	err = json.Unmarshal(segments[1], &etcdServerList)
	if err != nil {
		err = errors.WithStack(err)
		return err
	}

	num := len(etcdServerList) * len(registInfoList) * 2
	var retc = make(chan error, num)
	var rerr error

	for _, info := range registInfoList {
		for _, endpoints := range etcdServerList {
			k := strings.Join(endpoints, ",")

			cli, ok := pm.etcdclients[k]
			if !ok {
				cli, err = etcdclient.New(etcdclient.Config{
					Endpoints:            endpoints,
					AutoSyncInterval:     conf.EtcdSyncEndpointsTime,
					DialTimeout:          conf.EtcdDialTimeout,
					DialKeepAliveTime:    conf.EtcdDialKeepAliveTime,
					DialKeepAliveTimeout: conf.EtcdDialKeepAliveTimeout,
				})

				if err != nil {
					return err
				}

				pm.etcdclients[k] = cli
			}

			go Revoke(cli, info.larkString(), retc)
			go Revoke(cli, info.mcpString(), retc)
		}
	}

	for i := 0; i < num; i++ {
		select {
		case err := <-retc:
			if err != nil {
				if rerr == nil {
					rerr = err
				} else {
					rerr = errors.WithMessage(rerr, err.Error())
				}
			}
		case <-time.After(time.Second * 10):
			m := "keepalive wait response timeout"
			if rerr == nil {
				rerr = errors.New(m)
			} else {
				rerr = errors.WithMessage(rerr, m)
			}
		}
	}
	return rerr
}

func (pm *ProxyManager) Close() error {
	if atomic.CompareAndSwapUint32(&pm.status, runningFlag, closedFlag) {
		pm.lis.Close()
	}
	return nil
}

func (pm *ProxyManager) autoCheckExpireService() {
	tick := time.NewTicker(time.Second)
	for range tick.C {
		pm.checkExpireService()
	}
}

func (pm *ProxyManager) checkExpireService() {
	pm.Lock()
	defer pm.Unlock()

	for key, service := range pm.proxyMap {
		if time.Since(service.LastKeepAliveTime()) > time.Second*16 {
			if service.TryClose() {
				delete(pm.proxyMap, key)
			}
		}
	}
}

func (pm *ProxyManager) autoCheckBlockingTask() {
	tick := time.NewTicker(time.Millisecond * 500)
	for range tick.C {
		pm.checkBlockingTask()
	}
}

func (pm *ProxyManager) checkBlockingTask() {
	pm.Lock()
	defer pm.Unlock()

	for _, service := range pm.proxyMap {
		if n := len(service.TaskChan()); n > 5 {
			logger.Warnf("task is blocking, service:%s, blocking number:%d", service.ServiceName(), n)
		}
	}
}

func (pm *ProxyManager) BlockingTask() ([]string, []int) {
	pm.Lock()
	defer pm.Unlock()

	var names []string
	var nums []int
	for _, service := range pm.proxyMap {
		names = append(names, service.ServiceName())
		nums = append(nums, len(service.TaskChan()))
	}

	return names, nums
}
